package com.pw.study.flink.watermark;

import com.pw.study.flink.entities.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;


public class WatermarkSide {

    public static void main(String[] args) {
        wc2();
    }

    private static void wc2() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> stream = env
                .socketTextStream("hadoop111", 9999)
                .map(value -> {
                    String[] data = value.split(",");
                    return new WaterSensor(
                            data[0],
                            Long.valueOf(data[1]),
                            Integer.valueOf(data[2])
                    );
                });
        //2.策略设置
        WatermarkStrategy<WaterSensor> strategy = WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((SerializableTimestampAssigner<WaterSensor>) (sensor, l) -> sensor.getTs() * 1000)
                .withIdleness(Duration.ofSeconds(10));// 防止数据倾斜带来的水印不更新

        OutputTag<WaterSensor> other = new OutputTag<WaterSensor>("other") {
        };
        stream.process(new ProcessFunction<WaterSensor, WaterSensor>() {
            @Override
            public void processElement(WaterSensor ws, Context ctx, Collector<WaterSensor> out) throws Exception {

                if (ws.getId().equals("main")) {
                    System.out.println("z这是主流。。。。。");
                    out.collect(ws);
                } else if (ws.getId().equals("side")) {
                    System.out.println("这是side。。。。。");
                    ctx.output(new OutputTag<WaterSensor>("side") {
                    }, ws);
                } else {
                    System.out.println("other.......");
                    ctx.output(other, ws);
                }
            }
        });
        stream.print("main");
        stream.getSideOutput(other).print("other");
        stream.getSideOutput(new OutputTag<WaterSensor>("side"){}).print("side");

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
